package com.ctsi.canal.starter.handler;

import com.alibaba.otter.canal.protocol.Message;
import com.ctsi.canal.starter.context.CanalContext;
import com.ctsi.canal.starter.model.CanalModel;
import com.ctsi.canal.starter.util.HandlerUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;

import static com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import static com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import static com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import static com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import static com.alibaba.otter.canal.protocol.CanalEntry.RowData;

/**
 * @author ajh
 * @date 2019/10/24 15:24
 */
public abstract class AbstractMessageHandler implements MessageHandler<Message> {

    private Logger log = LoggerFactory.getLogger(AbstractMessageHandler.class);

    private Map<String, EntryHandler> tableHandlerMap;

    private RowDataHandler<RowData> rowDataHandler;


    public AbstractMessageHandler(List<? extends EntryHandler> entryHandlers, RowDataHandler<RowData> rowDataHandler) {
        this.tableHandlerMap = HandlerUtil.getTableHandlerMap(entryHandlers);
        this.rowDataHandler = rowDataHandler;
    }

    /**
     * 处理消息
     *
     * @param message message
     */
    @Override
    public void handleMessage(Message message) {
        List<Entry> entries = message.getEntries();
        for (Entry entry : entries) {

            if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN
                    || entry.getEntryType() == EntryType.TRANSACTIONEND) {
                //过滤事务消息
                continue;
            }

            if (entry.getEntryType().equals(EntryType.ROWDATA)) {

                try {
                    EntryHandler<?> entryHandler = HandlerUtil.getEntryHandler(tableHandlerMap, entry.getHeader().getTableName());
                    if (entryHandler != null) {
                        CanalModel model = CanalModel.Builder.builder()
                                .id(message.getId())
                                .table(entry.getHeader().getTableName())
                                .executeTime(entry.getHeader().getExecuteTime())
                                .schemaName(entry.getHeader().getSchemaName())
                                .build();
                        CanalContext.setModel(model);
                        RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
                        List<RowData> rowDataList = rowChange.getRowDatasList();
                        EventType eventType = rowChange.getEventType();
                        for (RowData rowData : rowDataList) {
                            rowDataHandler.handlerRowData(rowData, entryHandler, eventType);
                        }
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    throw new RuntimeException("parse event has an error , data:" + entry.toString(), e);
                } finally {
                    CanalContext.removeModel();
                }
            }
        }
    }

}
